// Copyright (C) 2008 Taylor L. Riche <riche@cs.utexas.edu>
//  
// This program is free software; you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation; either version 2 of the License, or
// (at your option) any later version.
// 
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.
// 
// You should have received a copy of the GNU General Public License
// along with this program; if not, write to the Free Software 
// Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
//

#include "Operator.h"
#include "ProcElement.h"
#include "Mutex.h"
#include "System.h"

namespace lagniappe
{

Operator::Operator(bool prof) : count(0), underConsideration(false), inProf(prof)
{
}

Operator::Operator(std::string s, Mutex * m, System * sys, bool prof) : name(s), count(0), mutex(m), system(sys), underConsideration(false), inProf(prof)
{
}

Operator::~Operator()
{
}

ProcElement * 
Operator::getProc(RCon * rcon)
{
/*	mutex->lock();
	assert(!procList.empty());
	assert(rcon->getRData() != NULL);
	ProcElement * retVal;
	if(count >= procList.size()) {count = 0;} // safety check
	assert(count < procList.size());
	retVal = procList[count++];
	assert(retVal != NULL);
	if(count >= procList.size()) {count = 0;}
	assert(count < procList.size());
	mutex->unlock();
	*/
	int size = procList.size();
	//mutex->lock();
	int rnum = (int)(size * (rand() / (RAND_MAX + 1.0)));
	//mutex->unlock();
	assert(rnum < size);
	return procList[rnum];
}

void
Operator::run(RCon * rcon) 
{
	assert(rcon != NULL);
	ProcElement * proc = this->getProc(rcon);
	assert(proc != NULL);
	if(!underConsideration && proc->isOverloaded(rcon)) {
		sendMonitorReq(MORE_RESOURCES, static_cast<void *>(proc), rcon->getProc());
	}
	// lookup correct source channel based on destination processor
	// assuming only one connection between processors at this point...
	// TODO put in choice of "fastest" or "fattest" channel right here as well
	ProcElement * srcProc = rcon->getProc();
	std::cerr << "Src Proc: " << srcProc->getName() << std::endl;
	std::cerr << "Dest Proc: " << proc->getName() << std::endl;
	CommChannel * chan = srcProc->lookupDestChan(proc);
	chan->put(rcon);
}

void
Operator::run(std::queue<RCon *> * rconQueue)
{
	assert(rconQueue != NULL);
	while(!rconQueue->empty()) {
		ProcElement * proc = this->getProc(rconQueue->front());
		assert(proc != NULL);
		if(!underConsideration && proc->isOverloaded(rconQueue->front())) {
			sendMonitorReq(MORE_RESOURCES, static_cast<void *>(proc), proc);
		}
		proc->processRequest(rconQueue->front());
		rconQueue->pop();
	}
	delete rconQueue;
		
}

void
Operator::LC_handleMonitor(RData * data)
{
	mutex->lock();
	MonitorResponse * response = static_cast<MonitorResponse *>(data);
	if(response->msgType == ADD_PROC) {
		// MON_DEBUG
		std::cerr << mon_debug(this, response->msgType) << "::" << response->respBody.addProc.newProc << std::endl;
		schedule(response->respBody.addProc.newProc);
		underConsideration = false;
	}
	else if(response->msgType == REVOKE_PROC) {
		// MON_DEBUG
		std::cerr << mon_debug(this, response->msgType) << std::endl;
		for(std::vector<ProcElement *>::iterator iter = procList.begin(); iter != procList.end(); ++iter) {
			if(*iter == response->respBody.revokeProc.revokedProc) {
				procList.erase(iter);
				break;
			}
		}
		this->reassignFlow(response->respBody.revokeProc.revokedProc);
	}
	else if(response->msgType == RESOURCES_EXHAUSTED) {
		// MON_DEBUG
		std::cerr << mon_debug(this, response->msgType) << std::endl;
		underConsideration = true;
	}
	//ProcElement * tempProc = lockedProc;
	lockedProc = NULL;
	mutex->unlock();
	//tempProc->unFreeze();
}

void 
Operator::schedule(ProcElement * p, bool initProc) 
{
	procList.push_back(p);
	if(initProc) {
		sendMonitorReq(INIT_PROC, static_cast<void *>(p), p);
	}
}

void
Operator::writeProfileInfo(std::ofstream & outfile)
{
	outfile << "<operator name=\"" << name << "\">" << std::endl;
	for(std::map<std::string, double>::iterator iter = profMap.begin(); iter != profMap.end(); iter++) {
		outfile << "\t<port name=\"" << iter->first << "\">" << iter->second / (double)PROF_SIZE << "</port>" << std::endl;
	}
	outfile << "</operator>" << std::endl;
}

void
Operator::profile(System * sys)
{
//	std::list<RCon *> * tempList = this->generateRequestList();
//	for(std::list<RCon *>::iterator iter = tempList->begin(); iter != tempList->end(); iter++) {
//		proc->profileLock();
//		for(unsigned int i = 0; i < PROF_SIZE; ++i) {
//			proc->putProfRCon(new RCon(*(*iter)));
//		}
//		proc->startTimer();
//		proc->profileNotify();
//		proc->profileWait();
//		this->profMap[(*iter)->getPort()] = proc->endTimer();
//		proc->profileUnlock();
//	}
	std::list<RCon *> * tempList = this->generateRequestList();
	for(std::list<RCon *>::iterator iter = tempList->begin(); iter != tempList->end(); iter++) {
		ProcElement * proc = sys->createProfilerResources();
		proc->init();
		profileRConEnqueue(proc, *iter);
		profileMeasure(proc, *iter);
		sys->destroyProfilerResources();
	}
	
}

void
Operator::profileRConEnqueue(ProcElement * proc, RCon * rcon)
{
	proc->profileLock();
	for(unsigned int i = 0; i < PROF_SIZE; ++i) {
		proc->putProfRCon(new RCon(*rcon));
	}
	proc->startTimer();
	proc->profileNotify();
	proc->profileUnlock();
}

void
Operator::profileMeasure(ProcElement * proc, RCon * rcon)
{
	proc->profileLock();
	while(!proc->profileDone()) {
		proc->profileWait();
	}
	this->profMap[rcon->getPort()] = proc->endTimer();
	proc->profileUnlock();
}

void 
Operator::sendMonitorReq(ReqType const & rt, void * v, ProcElement * proc) 
{
	if(name != "monitor") { 
	mutex->lock();
	underConsideration = true;
	//lockedProc = static_cast<ProcElement *>(v);
	lagniappe::CoreOp * op = destMap[MONITOR_OUT_PORT]->first;
	std::string port = destMap[MONITOR_OUT_PORT]->second;
	MonitorRequest * req = new MonitorRequest();
	req->msgType = rt;
	req->op = this; 
	if(rt == INIT_PROC) {
		assert(v != NULL);
		req->reqBody.initProc.proc = static_cast<ProcElement *>(v);
		underConsideration = false;
	}
	lagniappe::RCon * rc = new lagniappe::RCon(op, static_cast<RData *>(req), port, sizeof(*req), proc);
	mutex->unlock();
	op->run(rc);
	// MON_DEBUG
	std::cerr << mon_debug(this, rt) << std::endl;
	}
}

Mutex * 
Operator::getNewMutex() 
{
	return system->getNewMutex();
}

}
